import { Readable } from 'stream';

const END_SYMBOL = Symbol('STREAM_END');

type ReadResult<T> = 
  | { done: false; value: T }
  | { done: true; value?: never };

class StreamReader<T = Buffer> {
  private stream: Readable;
  private ended = false;
  private destroyed = false;
  private pending: null | {
    resolve: (item: { value: T | null; error?: never } | { error: Error }) => void;
    reject?: never;
  } = null;
  private bufferQueue: { data: T; bytes: number }[] = [];
  private totalBufferedBytes = 0;
  
  constructor(stream: Readable, private chunkSize = 16384) {
    if (!stream.readable || stream.destroyed) {
      throw new Error('Invalid readable stream');
    }

    this.stream = stream
      .on('readable', this.handleReadable)
      .on('end', this.handleEnd)
      .on('error', this.handleError)
      .on('close', this.destroy);

    // 初始读取填充缓冲区
    process.nextTick(this.fillBuffer);
  }

  // 高效填充缓冲区
  private fillBuffer = () => {
    if (this.ended || this.destroyed) return;

    let chunk: T | null;
    while ((chunk = this.stream.read(this.chunkSize) as T | null) !== null) {
      const chunkSize = this.getChunkSize(chunk);
      this.bufferQueue.push({ data: chunk, bytes: chunkSize });
      this.totalBufferedBytes += chunkSize;
      
      // 高水位线控制 (80% of highWaterMark)
      if (this.totalBufferedBytes > this.stream.readableHighWaterMark * 0.8) {
        this.stream.pause();
        break;
      }
    }

    this.resolvePending();
  };

  // 处理可读事件
  private handleReadable = () => {
    this.fillBuffer();
  };

  // 处理流结束
  private handleEnd = () => {
    if (this.ended) return;
    this.ended = true;
    this.resolvePending();
    this.destroy();
  };

  // 处理错误
  private handleError = (error: Error) => {
    if (this.destroyed) return;
    
    if (this.pending) {
      this.pending.resolve({ error });
      this.pending = null;
    }
    
    this.destroy();
  };

  // 解析待处理的读取请求
  private resolvePending() {
    if (!this.pending || this.bufferQueue.length === 0) return;

    const chunkObj = this.bufferQueue.shift();
    if (!chunkObj) return;
    
    this.totalBufferedBytes -= chunkObj.bytes;
    this.pending.resolve({ value: chunkObj.data });
    this.pending = null;
    
    // 恢复流处理
    if (this.stream.isPaused() && !this.ended) {
      this.stream.resume();
    }
  }

  // 获取数据块大小
  private getChunkSize(chunk: T): number {
    if (Buffer.isBuffer(chunk)) return chunk.length;
    if (typeof chunk === 'string') return Buffer.byteLength(chunk);
    if (ArrayBuffer.isView(chunk)) return chunk.byteLength;
    return 0;
  }

  /**
   * 拉取模式读取数据
   */
  public async read(): Promise<ReadResult<T>> {
    if (this.destroyed) throw new Error('StreamReader destroyed');
    if (this.ended && this.bufferQueue.length === 0) return { done: true };

    // 从缓冲区获取数据（如果可用）
    if (this.bufferQueue.length > 0) {
      const chunkObj = this.bufferQueue.shift()!;
      this.totalBufferedBytes -= chunkObj.bytes;
      return { done: false, value: chunkObj.data };
    }

    // 创建新的读取请求
    return new Promise<ReadResult<T>>((resolve, reject) => {
      this.pending = {
        resolve: (result) => {
          if ('error' in result) {
            reject(result.error);
          } else if (result.value === null) {
            resolve({ done: true });
          } else {
            resolve({ done: false, value: result.value });
          }
        }
      };

      // 触发可能的待处理数据
      process.nextTick(this.fillBuffer);
    });
  }

  /**
   * 自动流动模式
   */
  public async flow(
    callback: (chunk: T) => Promise<void>,
    parallel = 1
  ): Promise<void> {
    if (this.destroyed) return;

    try {
      const workers: Promise<void>[] = [];

      // 并行工作器
      for (let i = 0; i < parallel; i++) {
        workers.push(this.workerLoop(callback));
      }

      await Promise.all(workers);
    } finally {
      this.destroy();
    }
  }

  // 并行处理工作器
  private async workerLoop(callback: (chunk: T) => Promise<void>) {
    while (!this.ended || this.bufferQueue.length > 0) {
      const result = await this.read();
      
      if (result.done) break;
      await callback(result.value);
    }
  }

  /**
   * 异步迭代器
   */
  public async *[Symbol.asyncIterator](): AsyncGenerator<T, void, void> {
    while (true) {
      const result = await this.read();
      if (result.done) return;
      yield result.value;
    }
  }

  /**
   * 安全销毁流和读取器
   */
  public destroy = (): void => {
    if (this.destroyed) return;
    
    this.destroyed = true;
    this.ended = true;
    this.bufferQueue = [];
    this.totalBufferedBytes = 0;
    
    // 清理事件监听
    this.stream.removeListener('readable', this.handleReadable);
    this.stream.removeListener('end', this.handleEnd);
    this.stream.removeListener('error', this.handleError);
    this.stream.removeListener('close', this.destroy);
    
    // 销毁流
    if (!this.stream.destroyed) {
      this.stream.destroy();
    }
    
    // 解决待处理的Promise
    if (this.pending) {
      this.pending.resolve({ value: null });
      this.pending = null;
    }
  };
}

export default StreamReader;
